package com.zhang.flink.example;

import com.zhang.flink.bean.OrderEvent;
import com.zhang.flink.bean.ReceiptEvent;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @title: 实时对账
 * @author: zhang
 * @date: 2022/2/12 19:27
 */
public class OrderReceiptWithTimerState {
    public static void main(String[] args) throws Exception {
        // todo 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // todo 读取支付数据流数据
        SingleOutputStreamOperator<OrderEvent> orderEventDS = env
                .readTextFile("src/main/resources/OrderLog.csv")
                //转换为pojo class
                .map(new MapFunction<String, OrderEvent>() {
                    @Override
                    public OrderEvent map(String value) throws Exception {
                        String[] fields = value.split(",");
                        return new OrderEvent(
                                fields[0], fields[1], fields[2],
                                Long.parseLong(fields[3]) * 1000L
                        );
                    }
                })
                //过滤出支付数据
                .filter(r -> "pay".equals(r.type))
                //分配时间戳和水位线
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<OrderEvent>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<OrderEvent>() {
                                    @Override
                                    public long extractTimestamp(OrderEvent element, long recordTimestamp) {
                                        return element.ts;
                                    }
                                })
                );

        //todo 读取到账数据
        SingleOutputStreamOperator<ReceiptEvent> receiptEventDS = env
                .readTextFile("src/main/resources/ReceiptLog.csv")
                .map(new MapFunction<String, ReceiptEvent>() {
                    @Override
                    public ReceiptEvent map(String value) throws Exception {
                        String[] fields = value.split(",");
                        return new ReceiptEvent(
                                fields[0], fields[1],
                                Long.parseLong(fields[2]) * 1000L
                        );
                    }
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<ReceiptEvent>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<ReceiptEvent>() {
                                    @Override
                                    public long extractTimestamp(ReceiptEvent element, long recordTimestamp) {
                                        return element.ts;
                                    }
                                })
                );

        //todo 对账逻辑
        SingleOutputStreamOperator<Tuple2<OrderEvent, ReceiptEvent>> result = orderEventDS
                .keyBy(r -> r.payId)
                .connect(receiptEventDS.keyBy(r -> r.payId))
                .process(new MatchOrder());

        // todo 输出主流对账成功数据和提取侧输出流对账失败数据
        result.print("对账成功：");

        result.getSideOutput(new OutputTag<String>("no-receipt") {
        }).print("no-receipt:");

        result.getSideOutput(new OutputTag<String>("no-order") {
        }).print("no-order:");
        // todo 启动任务
        env.execute();
    }

    public static class MatchOrder extends CoProcessFunction<OrderEvent, ReceiptEvent, Tuple2<OrderEvent, ReceiptEvent>> {
        //定义支付状态变量
        private ValueState<OrderEvent> orderEventState;
        //定义到账状态变量
        private ValueState<ReceiptEvent> receiptEventState;
        //定义定时器时间状态变量
        private ValueState<Long> timerTs;

        @Override
        public void open(Configuration parameters) throws Exception {
            orderEventState = getRuntimeContext().getState(
                    new ValueStateDescriptor<OrderEvent>(
                            "order-state", Types.POJO(OrderEvent.class)
                    )
            );
            receiptEventState = getRuntimeContext().getState(
                    new ValueStateDescriptor<ReceiptEvent>(
                            "receipt-state", Types.POJO(ReceiptEvent.class)
                    )
            );
            timerTs = getRuntimeContext().getState(
                    new ValueStateDescriptor<Long>(
                            "timer-Ts", Types.LONG
                    )
            );
        }

        @Override
        public void processElement1(OrderEvent value, CoProcessFunction<OrderEvent, ReceiptEvent, Tuple2<OrderEvent, ReceiptEvent>>.Context ctx, Collector<Tuple2<OrderEvent, ReceiptEvent>> out) throws Exception {
            if (receiptEventState.value() == null) {
                //支付数据先到达，此时到账数据还未到达：
                //1 将支付数据存入状态
                orderEventState.update(value);
                //2 注册定时器等待到账数据10s
                long tenSecondLater = value.ts + 10 * 1000L;
                ctx.timerService().registerEventTimeTimer(tenSecondLater);
                //3 将定时器时间写入时间状态变量（用于删除定时器）
                timerTs.update(tenSecondLater);
            } else {
                //支付数据到的时候，到账数据已经到达，对账成功正常输出
                out.collect(Tuple2.of(value, receiptEventState.value()));
                //1 对账成功删除到账数据注册的定时器
                ctx.timerService().deleteEventTimeTimer(timerTs.value());
                //2 清空到账状态
                receiptEventState.clear();
                //3 清空定时器时间状态
                timerTs.clear();
            }
        }

        @Override
        public void processElement2(ReceiptEvent value, CoProcessFunction<OrderEvent, ReceiptEvent, Tuple2<OrderEvent, ReceiptEvent>>.Context ctx, Collector<Tuple2<OrderEvent, ReceiptEvent>> out) throws Exception {
            if (orderEventState.value() == null) {
                //到账数据先到达，此时支付数据还未到达；
                //1 将到账数据存入状态
                receiptEventState.update(value);
                //2 注册5s后的定时器，等待支付数据5s
                long fiveSecondLater = value.ts + 5 * 1000L;
                ctx.timerService().registerEventTimeTimer(fiveSecondLater);
                //3 将定时器时间写入时间状态变量（用于删除定时器）
                timerTs.update(fiveSecondLater);
            } else {
                //到账数据在支付数据等待的时间范围内到来，对账成功正常输出
                out.collect(Tuple2.of(orderEventState.value(), value));
                //1 删除支付数据注册定时器
                ctx.timerService().deleteEventTimeTimer(timerTs.value());
                //2 清空支付状态
                orderEventState.clear();
                //3 清空定时器时间状态
                timerTs.clear();
            }
        }

        //有以上逻辑可以推出，只要执行到定时器方法，只有两种情况：
        // 1 支付数据不为空，到账数据为空（到账数据没有在支付数据等待时间范围到达）
        // 2 到账数据不为空，支付数据为空（支付数据没有在到账数据等待时间范围内到达）
        // 以上两种情况互斥
        @Override
        public void onTimer(long timestamp, CoProcessFunction<OrderEvent, ReceiptEvent, Tuple2<OrderEvent, ReceiptEvent>>.OnTimerContext ctx, Collector<Tuple2<OrderEvent, ReceiptEvent>> out) throws Exception {
            /**
             * 如果去掉else逻辑可以实现左外连接
             * 如果去掉if逻辑可以实现右外连接
             * if 和 else 同时存在全外连接
             */
            if (orderEventState.value() != null) {
                //支付数据不为空，到账数据为空
                //对账失败发送到侧输出流
                ctx.output(new OutputTag<String>("no-receipt") {
                }, orderEventState.value().payId + "只有支付数据没有到账数据！");
            } else {
                //到账数据不为空，支付数据为空
                //对账失败发送到侧输出流
                ctx.output(new OutputTag<String>("no-order") {
                }, receiptEventState.value().payId + "只有到账数据没有支付数据！");
            }
            //清空状态
            orderEventState.clear();
            receiptEventState.clear();
            timerTs.clear();
        }
    }
}
